package org.ns.learn.rpc.register;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.ns.learn.rpc.RpcServer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PreDestroy;
import java.util.concurrent.CountDownLatch;

/**
 * RPC zookeeper 注册
 */
@Configuration
public class RpcServerZookeeperRegister implements InitializingBean, ApplicationContextAware {
    @Bean
    @ConditionalOnMissingBean
    RpcServerRegisterConfig rpcServerRegisterConfig() {
        return new RpcServerRegisterConfig();
    };

    ApplicationContext applicationContext;
    private ZooKeeper zk;

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    private void startServer(int port) throws Exception {
        RpcServer rpcServer = new RpcServer(applicationContext);
        Thread thread = new Thread(()->{
            rpcServer.start(port);
        });
        thread.start();
        Thread.sleep(1000);
    }

    private void zookeeperRegister(RpcServerRegisterConfig rpcServerRegisterConfig) throws Exception{
        String serverAddress = rpcServerRegisterConfig.getRegisterServerAddress();
        CountDownLatch connectAwait = new CountDownLatch(1);
        zk = new ZooKeeper(serverAddress, 3000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    connectAwait.countDown();
                }
            }
        });
        connectAwait.await();

        String path = "/rpcserver";
        Stat stat = zk.exists(path, false);
        if (stat == null) {
            try {
                // 持久节点
                zk.create(path, "ok".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } catch (KeeperException.NodeExistsException e) {
                // 节点已经存在的异常
            }
        }
        path = path + "/" + rpcServerRegisterConfig.getName();
        stat = zk.exists(path, false);
        if (stat == null) {
            try {
                // 持久节点
                zk.create(path, "ok".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } catch (KeeperException.NodeExistsException e) {
                // 节点已经存在的异常
            }
        }

        String host = rpcServerRegisterConfig.getHost();
        int port = rpcServerRegisterConfig.getPort();
        String nodePath = path + "/" + host + ":" + port;
        try {
            zk.create(nodePath, "ok".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("服务器注册成功: " + nodePath);
        } catch (KeeperException.NodeExistsException e) {
            // 节点已经存在的异常
            System.err.println("服务器注册节点已经存在！注册失败！, paht: " + e.getPath());
        }
    }



    @Autowired(required = false)
    RpcServerRegisterConfig rpcServerRegisterConfig;

    @Override
    public void afterPropertiesSet() throws Exception {
        if (rpcServerRegisterConfig != null) {
            try {
                startServer(rpcServerRegisterConfig.getPort());
                zookeeperRegister(rpcServerRegisterConfig);
            } catch (Exception e) {
                System.err.println("服务器注册失败！！");
                throw e;
            }
        }
    }

    @PreDestroy
    public void destroy() throws Exception {
        System.out.println("关闭啦" + zk);
        String path = "/rpcserver";
        String name = rpcServerRegisterConfig.getName();
        String host = rpcServerRegisterConfig.getHost();
        int port = rpcServerRegisterConfig.getPort();
        String nodePath = path + "/" + name + "/" + host + ":" + port;
        zk.delete(nodePath, -1);
        zk.close();
    }
}
